package com.ctvit.breadking.news.core.solr.repository.impl;

import com.ctvit.breadking.news.core.solr.config.SolrConfig;
import com.ctvit.breadking.news.core.solr.function.Function;
import com.ctvit.breadking.news.core.solr.repository.SolrDocumentRepository;
import com.ctvit.breadking.news.core.utils.ThreadMdcUtils;
import com.ctvit.breadking.news.core.utils.ThreadPoolUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.math.NumberUtils;
import org.apache.solr.client.solrj.SolrQuery;
import org.apache.solr.client.solrj.SolrServerException;
import org.apache.solr.client.solrj.impl.HttpSolrClient;
import org.apache.solr.client.solrj.impl.LBHttpSolrClient;
import org.apache.solr.client.solrj.response.QueryResponse;
import org.apache.solr.client.solrj.response.UpdateResponse;
import org.apache.solr.common.SolrInputDocument;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

/**
 * Solr文档服务实现类
 *
 * @author heyingcheng
 * @date 2024/10/2
 */
@Slf4j
@Service
public class SolrDocumentRepositoryImpl implements SolrDocumentRepository {

    private final SolrConfig solrConfig;

    private static final ThreadPoolTaskExecutor SYNC_THREAD_POOL_TASK_EXECUTOR = ThreadPoolUtils.create20ThreadPoolTaskExecutor();

    protected SolrDocumentRepositoryImpl(SolrConfig solrConfig) {
        this.solrConfig = solrConfig;
    }

    private boolean update(String collection, String id, Function<HttpSolrClient, UpdateResponse> function) {
        List<CompletableFuture<Boolean>> completableFutures = new ArrayList<>();
        List<HttpSolrClient> httpSolrClients = solrConfig.buildHttpSolrClientList();
        for (HttpSolrClient httpSolrClient : httpSolrClients) {
            completableFutures.add(CompletableFuture.supplyAsync(ThreadMdcUtils.wrapSupplier(() -> update(collection, id, httpSolrClient, function)), SYNC_THREAD_POOL_TASK_EXECUTOR));
        }

        try {
            CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[0])).get();
            for (CompletableFuture<Boolean> completableFuture : completableFutures) {
                if (Objects.equals(Boolean.FALSE, completableFuture.get())) {
                    return false;
                }
            }
            return true;
        } catch (InterruptedException e) {
            log.error("更新Solr文档中断异常, collection:[{}], id:[{}]", collection, id, e);
            Thread.currentThread().interrupt();
        } catch (ExecutionException e) {
            log.error("更新Solr文档执行异常, collection:[{}], id:[{}]", collection, id, e);
        }
        return false;
    }

    private boolean update(String collection, String id, HttpSolrClient httpSolrClient, Function<HttpSolrClient, UpdateResponse> function) {
        String baseUrl = httpSolrClient.getBaseURL();
        log.info("存储Solr文档开始, baseURL:[{}], collection:[{}], id:[{}]", baseUrl, collection, id);
        try {
            UpdateResponse response = function.apply(httpSolrClient);
            int status = response.getStatus();
            if (status == NumberUtils.INTEGER_ZERO) {
                httpSolrClient.commit(collection);
                return true;
            } else {
                log.error("存储Solr文档失败, baseURL:[{}], collection:[{}], id:[{}]", baseUrl, collection, id);
                return false;
            }
        } catch (SolrServerException e) {
            log.error("Solr服务器通信/解析异常, baseURL:[{}], collection:[{}], id:[{}]", baseUrl, collection, id, e);
        } catch (IOException e) {
            log.error("IO异常, baseURL:[{}], collection:[{}], id:[{}]", baseUrl, collection, id, e);
        } finally {
            try {
                httpSolrClient.close();
            } catch (IOException e) {
                log.error("Solr客户端端口异常, baseURL:[{}], collection:[{}], id:[{}]", baseUrl, collection, id, e);
            }
        }
        return false;
    }

    @Override
    public boolean add(String collection, String id, SolrInputDocument document) {
        return update(collection, id, httpSolrClient -> httpSolrClient.add(collection, document));
    }

    @Override
    public boolean deleteById(String collection, String id) {
        return update(collection, id, httpSolrClient -> httpSolrClient.deleteById(collection, id));
    }

    @Override
    public QueryResponse query(LBHttpSolrClient lbHttpSolrClient, SolrQuery solrQuery) {
        try {
            log.info("solrQueryString:[{}]", solrQuery.toQueryString());
            return lbHttpSolrClient.query(solrQuery);
        } catch (SolrServerException e) {
            log.error("Solr服务器通信/解析异常", e);
        } catch (IOException e) {
            log.error("IO异常", e);
        } finally {
            if (Objects.nonNull(lbHttpSolrClient)) {
                lbHttpSolrClient.close();
            }
        }
        return null;
    }

}
